package plugin.voldemort.lookup;

import java.util.List;
import java.util.Map;

import org.eclipse.swt.widgets.Shell;
import org.pentaho.di.core.*;
import org.pentaho.di.core.database.DatabaseMeta; 
import org.pentaho.di.core.exception.*;
import org.pentaho.di.core.row.*;
import org.pentaho.di.core.variables.VariableSpace;
import org.pentaho.di.core.xml.XMLHandler;
import org.pentaho.di.i18n.BaseMessages;
import org.pentaho.di.repository.*;
import org.pentaho.di.trans.*;
import org.pentaho.di.trans.step.*;
import org.w3c.dom.Node;

public class LookupStepMeta extends BaseStepMeta implements StepMetaInterface {

	private static Class<?> PKG = LookupStepMeta.class; // for i18n purposes
	
	
	private String voldemortHost;
	private String voldemortPort;

	public LookupStepMeta() {
		super(); 
	}
	
	// getters and setters for the step settings

	public String getVoldemortHost() {
		return voldemortHost;
	}

	public void setVoldemortHost(String voldemortHost) {
		this.voldemortHost = voldemortHost;
	}

	public String getVoldemortPort() {
		return voldemortPort;
	}

	public void setVoldemortPort(String voldemortPort) {
		this.voldemortPort = voldemortPort;
	}

	// set sensible defaults for a new step
	public void setDefault() {
		voldemortHost = "";
		voldemortPort = " ";

		// default is to have no key lookup settings
		allocate(0);
		
	}	
	
	// helper method to allocate the arrays
	public void allocate(int nrkeys){
		
		
	}

	public Object clone() {
		
		// field by field copy is default
		LookupStepMeta retval = (LookupStepMeta) super.clone();
		return retval;
	}	
	
	public String getXML() throws KettleValueException {
		
		StringBuffer retval = new StringBuffer(150);
			
		retval.append("    ").append(XMLHandler.addTagValue("host", voldemortHost));
		retval.append("    ").append(XMLHandler.addTagValue("port", voldemortPort));
	
		return retval.toString();
	}
	
	public void loadXML(Node stepnode, List<DatabaseMeta> databases, Map<String, Counter> counters) throws KettleXMLException {

		try {
			
			voldemortHost = XMLHandler.getTagValue(stepnode, "host");
			voldemortPort = XMLHandler.getTagValue(stepnode, "port");
			
			int nrKeys = XMLHandler.countNodes(stepnode, "lookup"); 
			allocate(nrKeys);
			
		} catch (Exception e) {
			throw new KettleXMLException("Template Plugin Unable to read step info from XML node", e);
		}

	}	


	public void readRep(Repository rep, ObjectId id_step, List<DatabaseMeta> databases, Map<String, Counter> counters) throws KettleException {
		try
		{
			voldemortHost = rep.getStepAttributeString(id_step, "host");
			voldemortPort = rep.getStepAttributeString(id_step, "port");
			//voldemortStore = rep.getStepAttributeString(id_step, "store");
			
			int nrKeys   = rep.countNrStepAttributes(id_step, "lookup_keyfield");
			allocate(nrKeys);
					}
		catch(Exception e)
		{
			throw new KettleException(BaseMessages.getString(PKG, "VoldemortStep.Exception.UnexpectedErrorInReadingStepInfo"), e);
		}
	}

	public void saveRep(Repository rep, ObjectId id_transformation, ObjectId id_step) throws KettleException
	{
		try
		{
			rep.saveStepAttribute(id_transformation, id_step, "host", voldemortHost); 
			rep.saveStepAttribute(id_transformation, id_step, "port", voldemortPort); 

		}
		catch(Exception e)
		{
			throw new KettleException(BaseMessages.getString(PKG, "VoldemortStep.Exception.UnableToSaveStepInfoToRepository")+id_step, e); 
		}
	}	
	


	public void check(List<CheckResultInterface> remarks, TransMeta transmeta, StepMeta stepMeta, RowMetaInterface prev, String input[], String output[], RowMetaInterface info) {
		CheckResult cr;

		// See if we have input streams leading to this step!
		if (input.length > 0) {
			cr = new CheckResult(CheckResult.TYPE_RESULT_OK, BaseMessages.getString(PKG, "VoldemortStep.Check.StepIsReceivingInfoFromOtherSteps"), stepMeta);
			remarks.add(cr);
		} else {
			cr = new CheckResult(CheckResult.TYPE_RESULT_ERROR, BaseMessages.getString(PKG, "VoldemortStep.Check.NoInputReceivedFromOtherSteps"), stepMeta);
			remarks.add(cr);
		}	
		
		// also check that each expected key fields are acually coming
		if (prev!=null && prev.size()>0)
		{
			boolean first=true;
			String error_message = ""; 
			boolean error_found = false;
			
			if (error_found)
			{
				cr = new CheckResult(CheckResultInterface.TYPE_RESULT_ERROR, error_message, stepMeta);
			}
			else
			{
				cr = new CheckResult(CheckResultInterface.TYPE_RESULT_OK, BaseMessages.getString(PKG, "VoldemortStep.Check.AllFieldsFoundInInput"), stepMeta);
			}
			remarks.add(cr);
		}
		else
		{
			String error_message=BaseMessages.getString(PKG, "VoldemortStep.Check.CouldNotReadFromPreviousSteps")+Const.CR;
			cr = new CheckResult(CheckResultInterface.TYPE_RESULT_ERROR, error_message, stepMeta);
			remarks.add(cr);
		}		
    	
	}

	public StepDialogInterface getDialog(Shell shell, StepMetaInterface meta, TransMeta transMeta, String name) {
		return new LookupStepDialog(shell, meta, transMeta, name);
	}

	public StepInterface getStep(StepMeta stepMeta, StepDataInterface stepDataInterface, int cnr, TransMeta transMeta, Trans disp) {
		return new LookupStep(stepMeta, stepDataInterface, cnr, transMeta, disp);
	}

	public StepDataInterface getStepData() {
		return new LookupStepData();
	}
}
